package com.atguigu.gmall.realtime.test;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: Flink01_Sql
 * Description:
 *
 * @Create 2024/4/19 16:16
 */
public class Flink01_Sql {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 设置状态过期时间
        //tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        SingleOutputStreamOperator<Emp> empDS = env.socketTextStream("hadoop102", 8888)
                .map(
                        new MapFunction<String, Emp>() {
                            @Override
                            public Emp map(String value) throws Exception {
                                String[] fieldArr = value.split(",");
                                return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                            }
                        }
                );
        tableEnv.createTemporaryView("emp",empDS);

        SingleOutputStreamOperator<Dept> deptDS = env.socketTextStream("hadoop102", 8889)
                .map(
                        new MapFunction<String, Dept>() {
                            @Override
                            public Dept map(String value) throws Exception {
                                String[] fieldArr = value.split(",");
                                return new Dept(Integer.valueOf(fieldArr[0]), fieldArr[1], Long.valueOf(fieldArr[2]));
                            }
                        }
                );
        tableEnv.createTemporaryView("dept",deptDS);

        //TODO 内连接
        /*
        如果使用普通的内外连接，底层会为参与连接的两张表各自维护一个状态，默认情况下，永远不失效

        左表状态过期时间：OnCreateAndWrite
        右表状态过期时间：OnCreateAndWrite

        生产环境下一定要设置状态过期时间
        tableEnv.executeSql("select * from emp e join dept d on e.deptno = d.deptno").print();
        */

        //TODO 左外连接
        /*
        如果左外连接左表数据先到，右表后到，那会产生如下三条消息
         +I   左表  null
         -D   左表  null
         +I   左表  右表
        这样的动态表转换的流称为回撤流

        左表状态过期时间：OnReadAndWrite
        右表状态过期时间：OnCreateAndWrite
        */

        //tableEnv.executeSql("select * from emp e left join dept d on e.deptno = d.deptno").print();

        //TODO 右外连接（参左外连接）

        //TODO 全外连接
        /*
        左表状态过期时间：OnReadAndWrite
        右表状态过期时间：OnReadAndWrite

        tableEnv.executeSql("select * from emp e full join dept d on e.deptno = d.deptno").print();
        */

        //TODO 将左外连接的结果写到kafka
        //如果左表数据先到，右表数据后到，kafka主题会收到3条消息
        //  左表  null
        //    null
        //  左表  右表
        //FlinkSQL读取kafka数据，会对空消息自动进行处理
        //FlinkAPI读取kafka数据，如果使用默认的SimpleStringSchema进行反序列化，是处理不了空消息的，底层要求notnull。需要自定义反序列化器
        //另外，剩下的其他两条数据属于重复数据，DWS读取DWD数据的时候，如果存在重复需要去重
        //1.创建动态表 和要写入的主题进行映射

        tableEnv.executeSql("CREATE TABLE emp_dept (\n" +
                "  empno integer,\n" +
                "  ename STRINLG,\n" +
                "  deptno integer,\n" +
                "  dname STRING,\n" +
                "  PRIMARY KEY (empno) NOT ENFORCLED\n" +
                ") WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = 'first',\n" +
                "  'properties.bootstrap.servers' = 'hadoop103:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")");

        //2.写入
        tableEnv.executeSql("insert into emp_dept select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno");


    }
}
